//
// Created by changlei on 2022/4/7.
//

import Foundation
import RxSwift
import Dio
import Alamofire

/// Default instance of unknown error
public let RxDioUnknownError = NSError(domain: "RxDioDomain", code: -1, userInfo: nil)

// MARK: Convenience functions

/**
 Creates a NSMutableURLRequest using all necessary parameters.
 
 - parameter method: Dio method object
 - parameter url: An object adopting `URLConvertible`
 - parameter parameters: A dictionary containing all necessary options
 - parameter encoding: The kind of encoding used to process parameters
 - parameter header: A dictionary containing all the additional headers
 - returns: An instance of `NSMutableURLRequest`
 */

public func urlRequest(
    _ convertible: URLConvertible,
    method: HTTPMethod,
    parameters: Parameters? = nil,
    encoding: ParameterEncoding = URLEncoding.default,
    headers: HTTPHeaders? = nil
) throws -> Foundation.URLRequest {
    var mutableURLRequest = Foundation.URLRequest(url: try convertible.asURL())
    mutableURLRequest.httpMethod = method.rawValue
    
    if let headers = headers {
        for header in headers {
            mutableURLRequest.setValue(header.value, forHTTPHeaderField: header.name)
        }
    }
    
    if let parameters = parameters {
        mutableURLRequest = try encoding.encode(mutableURLRequest, with: parameters)
    }
    
    return mutableURLRequest
}

// MARK: Request

/**
 Creates an observable of the generated `Request`.
 
 - parameter method: Dio method object
 - parameter url: An object adopting `URLConvertible`
 - parameter parameters: A dictionary containing all necessary options
 - parameter encoding: The kind of encoding used to process parameters
 - parameter header: A dictionary containing all the additional headers
 - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
 - returns: An observable of a the `Request`
 */
public func request(
    _ convertible: URLConvertible,
    method: HTTPMethod,
    parameters: Parameters? = nil,
    encoding: ParameterEncoding = URLEncoding.default,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    requestModifier: Dio.RequestModifier? = nil
) -> Observable<DioDataRequest> {
    Dio.default.rx.request(
        convertible,
        method: method,
        parameters: parameters,
        encoding: encoding,
        headers: headers,
        interceptor: interceptor,
        requestModifier: requestModifier
    )
}

public func request<Parameters: Encodable>(
    _ convertible: URLConvertible,
    method: HTTPMethod = .get,
    parameters: Parameters? = nil,
    encoder: ParameterEncoder = URLEncodedFormParameterEncoder.default,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    requestModifier: Dio.RequestModifier? = nil
) -> Observable<DioDataRequest> {
    Dio.default.rx.request(
        convertible,
        method: method,
        parameters: parameters,
        encoder: encoder,
        headers: headers,
        interceptor: interceptor,
        requestModifier: requestModifier
    )
}

/**
 Creates an observable of the generated `Request`.
 
 - parameter urlRequest: An object adopting `URLRequestConvertible`
 - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
 - returns: An observable of a the `Request`
 */
public func request(
    _ convertible: URLRequestConvertible,
    interceptor: RequestInterceptor? = nil
) -> Observable<DioDataRequest> {
    Dio.default.rx.request(convertible, interceptor: interceptor)
}

public func request(
    _ convertible: URLConvertible,
    method: HTTPMethod,
    parameters: Parameters? = nil,
    encoding: ParameterEncoding = URLEncoding.default,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    requestModifier: Dio.RequestModifier? = nil
) -> DioDataRequest {
    Dio.default.rx.request(
        convertible,
        method: method,
        parameters: parameters,
        encoding: encoding,
        headers: headers,
        interceptor: interceptor,
        requestModifier: requestModifier
    )
}

public func request<Parameters: Encodable>(
    _ convertible: URLConvertible,
    method: HTTPMethod = .get,
    parameters: Parameters? = nil,
    encoder: ParameterEncoder = URLEncodedFormParameterEncoder.default,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    requestModifier: Dio.RequestModifier? = nil
) -> DioDataRequest {
    Dio.default.rx.request(
        convertible,
        method: method,
        parameters: parameters,
        encoder: encoder,
        headers: headers,
        interceptor: interceptor,
        requestModifier: requestModifier
    )
}

public func request(
    _ convertible: URLRequestConvertible,
    interceptor: RequestInterceptor? = nil
) -> DioDataRequest {
    Dio.default.rx.request(convertible, interceptor: interceptor)
}

// MARK: response

/**
 Creates an observable of the `NSHTTPURLResponse` instance.
 
 - parameter method: Dio method object
 - parameter url: An object adopting `URLConvertible`
 - parameter parameters: A dictionary containing all necessary options
 - parameter encoding: The kind of encoding used to process parameters
 - parameter header: A dictionary containing all the additional headers
 - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
 - returns: An observable of `NSHTTPURLResponse`
 */
public func requestResponse(
    _ convertible: URLConvertible,
    method: HTTPMethod,
    parameters: Parameters? = nil,
    encoding: ParameterEncoding = URLEncoding.default,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    requestModifier: Dio.RequestModifier? = nil
) -> Observable<HTTPURLResponse> {
    Dio.default.rx.response(
        convertible,
        method: method,
        parameters: parameters,
        encoding: encoding,
        headers: headers,
        interceptor: interceptor,
        requestModifier: requestModifier
    )
}

/**
 Creates an observable of the `NSHTTPURLResponse` instance.
 
 - parameter urlRequest: An object adopting `URLRequestConvertible`
 - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
 - returns: An observable of `NSHTTPURLResponse`
 */
public func requestResponse(
    _ convertible: URLRequestConvertible,
    interceptor: RequestInterceptor? = nil
) -> Observable<HTTPURLResponse> {
    request(convertible, interceptor: interceptor).flatMap {
        $0.rx.response()
    }
}

/**
 Creates an observable of the returned data.
 
 - parameter method: Dio method object
 - parameter url: An object adopting `URLConvertible`
 - parameter parameters: A dictionary containing all necessary options
 - parameter encoding: The kind of encoding used to process parameters
 - parameter header: A dictionary containing all the additional headers
 - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
 - returns: An observable of `NSHTTPURLResponse`
 */
public func response(
    _ convertible: URLConvertible,
    method: HTTPMethod,
    parameters: Parameters? = nil,
    encoding: ParameterEncoding = URLEncoding.default,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    requestModifier: Dio.RequestModifier? = nil
) -> Observable<HTTPURLResponse> {
    Dio.default.rx.response(
        convertible,
        method: method,
        parameters: parameters,
        encoding: encoding,
        headers: headers,
        interceptor: interceptor,
        requestModifier: requestModifier
    )
}

// MARK: data

/**
 Creates an observable of the `(NSHTTPURLResponse, NSData)` instance.
 
 - parameter method: Dio method object
 - parameter url: An object adopting `URLConvertible`
 - parameter parameters: A dictionary containing all necessary options
 - parameter encoding: The kind of encoding used to process parameters
 - parameter header: A dictionary containing all the additional headers
 - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
 - returns: An observable of a tuple containing `(NSHTTPURLResponse, NSData)`
 */
public func requestData(
    _ convertible: URLConvertible,
    method: HTTPMethod,
    parameters: Parameters? = nil,
    encoding: ParameterEncoding = URLEncoding.default,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    requestModifier: Dio.RequestModifier? = nil
) -> Observable<(HTTPURLResponse, Data)> {
    Dio.default.rx.responseData(
        convertible,
        method: method,
        parameters: parameters,
        encoding: encoding,
        headers: headers,
        interceptor: interceptor,
        requestModifier: requestModifier
    )
}

/**
 Creates an observable of the `(NSHTTPURLResponse, NSData)` instance.
 
 - parameter urlRequest: An object adopting `URLRequestConvertible`
 - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
 - returns: An observable of a tuple containing `(NSHTTPURLResponse, NSData)`
 */
public func requestData(
    _ convertible: URLRequestConvertible,
    interceptor: RequestInterceptor? = nil
) -> Observable<(HTTPURLResponse, Data)> {
    request(convertible, interceptor: interceptor).flatMap {
        $0.rx.responseData()
    }
}

/**
 Creates an observable of the returned data.
 
 - parameter method: Dio method object
 - parameter url: An object adopting `URLConvertible`
 - parameter parameters: A dictionary containing all necessary options
 - parameter encoding: The kind of encoding used to process parameters
 - parameter header: A dictionary containing all the additional headers
 - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
 - returns: An observable of `NSData`
 */
public func data(
    _ convertible: URLConvertible,
    method: HTTPMethod,
    parameters: Parameters? = nil,
    encoding: ParameterEncoding = URLEncoding.default,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    requestModifier: Dio.RequestModifier? = nil
) -> Observable<Data> {
    Dio.default.rx.data(
        convertible,
        method: method,
        parameters: parameters,
        encoding: encoding,
        headers: headers,
        interceptor: interceptor,
        requestModifier: requestModifier
    )
}

// MARK: string

/**
 Creates an observable of the returned decoded string and response.
 
 - parameter method: Dio method object
 - parameter url: An object adopting `URLConvertible`
 - parameter parameters: A dictionary containing all necessary options
 - parameter encoding: The kind of encoding used to process parameters
 - parameter header: A dictionary containing all the additional headers
 - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
 - returns: An observable of the tuple `(NSHTTPURLResponse, String)`
 */
public func requestString(
    _ convertible: URLConvertible,
    method: HTTPMethod,
    parameters: Parameters? = nil,
    encoding: ParameterEncoding = URLEncoding.default,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    requestModifier: Dio.RequestModifier? = nil
) -> Observable<(HTTPURLResponse, String)> {
    Dio.default.rx.responseString(
        convertible,
        method: method,
        parameters: parameters,
        encoding: encoding,
        headers: headers,
        interceptor: interceptor,
        requestModifier: requestModifier
    )
}

/**
 Creates an observable of the returned decoded string and response.
 
 - parameter urlRequest: An object adopting `URLRequestConvertible`
 - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
 - returns: An observable of the tuple `(NSHTTPURLResponse, String)`
 */
public func requestString(
    _ convertible: URLRequestConvertible,
    interceptor: RequestInterceptor? = nil
) -> Observable<(HTTPURLResponse, String)> {
    request(convertible, interceptor: interceptor).flatMap {
        $0.rx.responseString()
    }
}

/**
 Creates an observable of the returned decoded string.
 
 - parameter method: Dio method object
 - parameter url: An object adopting `URLConvertible`
 - parameter parameters: A dictionary containing all necessary options
 - parameter encoding: The kind of encoding used to process parameters
 - parameter header: A dictionary containing all the additional headers
 - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
 - returns: An observable of `String`
 */
public func string(
    _ convertible: URLConvertible,
    method: HTTPMethod,
    parameters: Parameters? = nil,
    encoding: ParameterEncoding = URLEncoding.default,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    requestModifier: Dio.RequestModifier? = nil
) -> Observable<String> {
    Dio.default.rx.string(
        convertible,
        method: method,
        parameters: parameters,
        encoding: encoding,
        headers: headers,
        interceptor: interceptor,
        requestModifier: requestModifier
    )
}

// MARK: JSON

/**
 Creates an observable of the returned decoded JSON as `AnyObject` and the response.
 
 - parameter method: Dio method object
 - parameter url: An object adopting `URLConvertible`
 - parameter parameters: A dictionary containing all necessary options
 - parameter encoding: The kind of encoding used to process parameters
 - parameter header: A dictionary containing all the additional headers
 - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
 - returns: An observable of the tuple `(NSHTTPURLResponse, AnyObject)`
 */
@available(*, deprecated, message: "requestJSON deprecated and will be removed in Dio 6. Use requestDecodable instead.")
public func requestJSON(
    _ convertible: URLConvertible,
    method: HTTPMethod,
    parameters: Parameters? = nil,
    encoding: ParameterEncoding = URLEncoding.default,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    requestModifier: Dio.RequestModifier? = nil
) -> Observable<(HTTPURLResponse, Any)> {
    Dio.default.rx.responseJSON(
        convertible,
        method: method,
        parameters: parameters,
        encoding: encoding,
        headers: headers,
        interceptor: interceptor,
        requestModifier: requestModifier
    )
}

/**
 Creates an observable of the returned decoded JSON as `AnyObject` and the response.
 
 - parameter urlRequest: An object adopting `URLRequestConvertible`
 - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
 - returns: An observable of the tuple `(NSHTTPURLResponse, AnyObject)`
 */
@available(*, deprecated, message: "requestJSON deprecated and will be removed in Dio 6. Use requestDecodable instead.")
public func requestJSON(
    _ convertible: URLRequestConvertible,
    interceptor: RequestInterceptor? = nil
) -> Observable<(HTTPURLResponse, Any)> {
    request(convertible, interceptor: interceptor).flatMap {
        $0.rx.responseJSON()
    }
}

/**
 Creates an observable of the returned decoded JSON.
 
 - parameter method: Dio method object
 - parameter url: An object adopting `URLConvertible`
 - parameter parameters: A dictionary containing all necessary options
 - parameter encoding: The kind of encoding used to process parameters
 - parameter header: A dictionary containing all the additional headers
 - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
 - returns: An observable of the decoded JSON as `Any`
 */
@available(*, deprecated, message: "json deprecated and will be removed in Dio 6. Use decodable instead.")
public func json(
    _ convertible: URLConvertible,
    method: HTTPMethod,
    parameters: Parameters? = nil,
    encoding: ParameterEncoding = URLEncoding.default,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    requestModifier: Dio.RequestModifier? = nil
) -> Observable<Any> {
    Dio.default.rx.json(
        convertible,
        method: method,
        parameters: parameters,
        encoding: encoding,
        headers: headers,
        interceptor: interceptor,
        requestModifier: requestModifier
    )
}

// MARK: Decodable

/**
 Creates an observable of the returned decoded Decodable as `T` and the response.
 
 - parameter method: Dio method object
 - parameter url: An object adopting `URLConvertible`
 - parameter parameters: A dictionary containing all necessary options
 - parameter encoding: The kind of encoding used to process parameters
 - parameter header: A dictionary containing all the additional headers
 - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
 - returns: An observable of the tuple `(NSHTTPURLResponse, T)`
 */
public func requestDecodable<T: Decodable>(
    _ convertible: URLConvertible,
    of type: T.Type = T.self,
    method: HTTPMethod,
    parameters: Parameters? = nil,
    encoding: ParameterEncoding = URLEncoding.default,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    requestModifier: Dio.RequestModifier? = nil
) -> Observable<(HTTPURLResponse, T)> {
    Dio.default.rx.responseDecodable(
        convertible,
        of: type,
        method: method,
        parameters: parameters,
        encoding: encoding,
        headers: headers,
        interceptor: interceptor,
        requestModifier: requestModifier
    )
}

/**
 Creates an observable of the returned decoded Decodable as `T` and the response.
 
 - parameter urlRequest: An object adopting `URLRequestConvertible`
 - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
 - returns: An observable of the tuple `(NSHTTPURLResponse, T)`
 */
public func requestDecodable<T: Decodable>(
    _ convertible: URLRequestConvertible,
    of type: T.Type = T.self,
    interceptor: RequestInterceptor? = nil
) -> Observable<(HTTPURLResponse, T)> {
    request(convertible, interceptor: interceptor).flatMap {
        $0.rx.responseDecodable(of: type)
    }
}

/**
 Creates an observable of the returned decoded Decodable.
 
 - parameter method: Dio method object
 - parameter url: An object adopting `URLConvertible`
 - parameter parameters: A dictionary containing all necessary options
 - parameter encoding: The kind of encoding used to process parameters
 - parameter header: A dictionary containing all the additional headers
 - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
 - returns: An observable of the decoded Decodable as `T`
 */
public func decodable<T: Decodable>(
    _ convertible: URLConvertible,
    of type: T.Type = T.self,
    method: HTTPMethod,
    parameters: Parameters? = nil,
    encoding: ParameterEncoding = URLEncoding.default,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    requestModifier: Dio.RequestModifier? = nil
) -> Observable<T> {
    Dio.default.rx.decodable(
        convertible,
        of: type,
        method: method,
        parameters: parameters,
        encoding: encoding,
        headers: headers,
        interceptor: interceptor,
        requestModifier: requestModifier
    )
}

#if swift(>=5.5)
@available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
public func requestResponse(
    _ convertible: URLConvertible,
    method: HTTPMethod,
    parameters: Parameters? = nil,
    encoding: ParameterEncoding = URLEncoding.default,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    requestModifier: Dio.RequestModifier? = nil
) async throws -> HTTPURLResponse {
    try await Dio.default.rx.response(
        convertible,
        method: method,
        parameters: parameters,
        encoding: encoding,
        headers: headers,
        interceptor: interceptor,
        requestModifier: requestModifier
    )
}

@available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
public func requestResponse(
    _ convertible: URLRequestConvertible,
    interceptor: RequestInterceptor? = nil
) async throws -> HTTPURLResponse {
    try await request(convertible, interceptor: interceptor).rx.response()
}

@available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
public func response(
    _ convertible: URLConvertible,
    method: HTTPMethod,
    parameters: Parameters? = nil,
    encoding: ParameterEncoding = URLEncoding.default,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    requestModifier: Dio.RequestModifier? = nil
) async throws -> HTTPURLResponse {
    try await Dio.default.rx.response(
        convertible,
        method: method,
        parameters: parameters,
        encoding: encoding,
        headers: headers,
        interceptor: interceptor,
        requestModifier: requestModifier
    )
}

@available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
public func requestData(
    _ convertible: URLConvertible,
    method: HTTPMethod,
    parameters: Parameters? = nil,
    encoding: ParameterEncoding = URLEncoding.default,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    requestModifier: Dio.RequestModifier? = nil
) async throws -> (HTTPURLResponse, Data) {
    try await Dio.default.rx.responseData(
        convertible,
        method: method,
        parameters: parameters,
        encoding: encoding,
        headers: headers,
        interceptor: interceptor,
        requestModifier: requestModifier
    )
}

@available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
public func requestData(
    _ convertible: URLRequestConvertible,
    interceptor: RequestInterceptor? = nil
) async throws -> (HTTPURLResponse, Data) {
    try await request(convertible, interceptor: interceptor).rx.responseData()
}

@available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
public func data(
    _ convertible: URLConvertible,
    method: HTTPMethod,
    parameters: Parameters? = nil,
    encoding: ParameterEncoding = URLEncoding.default,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    requestModifier: Dio.RequestModifier? = nil
) async throws -> Data {
    try await Dio.default.rx.data(
        convertible,
        method: method,
        parameters: parameters,
        encoding: encoding,
        headers: headers,
        interceptor: interceptor,
        requestModifier: requestModifier
    )
}

@available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
public func requestString(
    _ convertible: URLConvertible,
    method: HTTPMethod,
    parameters: Parameters? = nil,
    encoding: ParameterEncoding = URLEncoding.default,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    requestModifier: Dio.RequestModifier? = nil
) async throws -> (HTTPURLResponse, String) {
    try await Dio.default.rx.responseString(
        convertible,
        method: method,
        parameters: parameters,
        encoding: encoding,
        headers: headers,
        interceptor: interceptor,
        requestModifier: requestModifier
    )
}

@available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
public func requestString(
    _ convertible: URLRequestConvertible,
    interceptor: RequestInterceptor? = nil
) async throws -> (HTTPURLResponse, String) {
    try await request(convertible, interceptor: interceptor).rx.responseString()
}

@available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
public func string(
    _ convertible: URLConvertible,
    method: HTTPMethod,
    parameters: Parameters? = nil,
    encoding: ParameterEncoding = URLEncoding.default,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    requestModifier: Dio.RequestModifier? = nil
) async throws -> String {
    try await Dio.default.rx.string(
        convertible,
        method: method,
        parameters: parameters,
        encoding: encoding,
        headers: headers,
        interceptor: interceptor,
        requestModifier: requestModifier
    )
}

@available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
@available(*, deprecated, message: "requestJSON deprecated and will be removed in Dio 6. Use requestDecodable instead.")
public func requestJSON(
    _ convertible: URLConvertible,
    method: HTTPMethod,
    parameters: Parameters? = nil,
    encoding: ParameterEncoding = URLEncoding.default,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    requestModifier: Dio.RequestModifier? = nil
) async throws -> (HTTPURLResponse, Any) {
    try await Dio.default.rx.responseJSON(
        convertible,
        method: method,
        parameters: parameters,
        encoding: encoding,
        headers: headers,
        interceptor: interceptor,
        requestModifier: requestModifier
    )
}

@available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
@available(*, deprecated, message: "requestJSON deprecated and will be removed in Dio 6. Use requestDecodable instead.")
public func requestJSON(
    _ convertible: URLRequestConvertible,
    interceptor: RequestInterceptor? = nil
) async throws -> (HTTPURLResponse, Any) {
    try await request(convertible, interceptor: interceptor).rx.responseJSON()
}

@available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
@available(*, deprecated, message: "json deprecated and will be removed in Dio 6. Use decodable instead.")
public func json(
    _ convertible: URLConvertible,
    method: HTTPMethod,
    parameters: Parameters? = nil,
    encoding: ParameterEncoding = URLEncoding.default,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    requestModifier: Dio.RequestModifier? = nil
) async throws -> Any {
    try await Dio.default.rx.json(
        convertible,
        method: method,
        parameters: parameters,
        encoding: encoding,
        headers: headers,
        interceptor: interceptor,
        requestModifier: requestModifier
    )
}

@available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
public func requestDecodable<T: Decodable>(
    _ convertible: URLConvertible,
    of type: T.Type = T.self,
    method: HTTPMethod,
    parameters: Parameters? = nil,
    encoding: ParameterEncoding = URLEncoding.default,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    requestModifier: Dio.RequestModifier? = nil
) async throws -> (HTTPURLResponse, T) {
    try await Dio.default.rx.responseDecodable(
        convertible,
        of: type,
        method: method,
        parameters: parameters,
        encoding: encoding,
        headers: headers,
        interceptor: interceptor,
        requestModifier: requestModifier
    )
}

@available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
public func requestDecodable<T: Decodable>(
    _ convertible: URLRequestConvertible,
    of type: T.Type = T.self,
    interceptor: RequestInterceptor? = nil
) async throws -> (HTTPURLResponse, T) {
    try await request(convertible, interceptor: interceptor).rx.responseDecodable(of: type)
}

@available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
public func decodable<T: Decodable>(
    _ convertible: URLConvertible,
    of type: T.Type = T.self,
    method: HTTPMethod,
    parameters: Parameters? = nil,
    encoding: ParameterEncoding = URLEncoding.default,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    requestModifier: Dio.RequestModifier? = nil
) async throws -> T {
    try await Dio.default.rx.decodable(
        convertible,
        of: type,
        method: method,
        parameters: parameters,
        encoding: encoding,
        headers: headers,
        interceptor: interceptor,
        requestModifier: requestModifier
    )
}
#endif

// MARK: Upload

/**
 Returns an observable of a request using the shared manager instance to upload a specific file to a specified URL.
 The request is started immediately.
 
 - parameter file: An instance of NSURL holding the information of the local file.
 - parameter urlRequest: The request object to start the upload.
 - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
 - returns: The observable of `UploadRequest` for the created request.
 */
public func upload(
    _ file: URL,
    with convertible: URLRequestConvertible,
    interceptor: RequestInterceptor? = nil,
    fileManager: FileManager = .default
) -> Observable<DioUploadRequest> {
    Dio.default.rx.upload(
        file,
        with: convertible,
        interceptor: interceptor,
        fileManager: fileManager
    )
}

public func upload(
    _ file: URL,
    with convertible: URLRequestConvertible,
    interceptor: RequestInterceptor? = nil,
    fileManager: FileManager = .default
) -> Observable<RxProgress> {
    Dio.default.rx.upload(
        file,
        with: convertible,
        interceptor: interceptor,
        fileManager: fileManager
    )
}

/**
 Returns an observable of a request using the shared manager instance to upload a specific file to a specified URL.
 The request is started immediately.
 
 - parameter file: An instance of `URL` holding the information of the local file.
 - parameter url: An object adopting `URLConvertible`
 - parameter method: Dio method object
 - parameter headers: A `HTTPHeaders` containing all the additional headers
 - returns: The observable of `UploadRequest` for the created request.
 */
public func upload(
    _ file: URL,
    to convertible: URLConvertible,
    method: HTTPMethod = .post,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    fileManager: FileManager = .default,
    requestModifier: Dio.RequestModifier? = nil
) -> Observable<DioUploadRequest> {
    Dio.default.rx.upload(
        file,
        to: convertible,
        method: method,
        headers: headers,
        interceptor: interceptor,
        fileManager: fileManager,
        requestModifier: requestModifier
    )
}

/**
 Returns an observable of a request progress using the shared manager instance to upload a specific file to a specified URL.
 The request is started immediately.
 
 - parameter file: An instance of `URL` holding the information of the local file.
 - parameter url: An object adopting `URLConvertible`
 - parameter method: Dio method object
 - parameter headers: A `HTTPHeaders` containing all the additional headers
 - returns: The observable of `RxProgress` for the created request.
 */
public func upload(
    _ file: URL,
    to convertible: URLConvertible,
    method: HTTPMethod = .post,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    fileManager: FileManager = .default,
    requestModifier: Dio.RequestModifier? = nil
) -> Observable<RxProgress> {
    Dio.default.rx.upload(
        file,
        to: convertible,
        method: method,
        headers: headers,
        interceptor: interceptor,
        fileManager: fileManager,
        requestModifier: requestModifier
    )
}

/**
 Returns an observable of a request using the shared manager instance to upload any data to a specified URL.
 The request is started immediately.
 
 - parameter data: An instance of NSData holdint the data to upload.
 - parameter urlRequest: The request object to start the upload.
 - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
 - returns: The observable of `UploadRequest` for the created request.
 */
public func upload(
    _ data: Data,
    with convertible: URLRequestConvertible,
    interceptor: RequestInterceptor? = nil,
    fileManager: FileManager = .default
) -> Observable<DioUploadRequest> {
    Dio.default.rx.upload(
        data,
        with: convertible,
        interceptor: interceptor,
        fileManager: fileManager
    )
}

public func upload(
    _ data: Data,
    with convertible: URLRequestConvertible,
    interceptor: RequestInterceptor? = nil,
    fileManager: FileManager = .default
) -> Observable<RxProgress> {
    Dio.default.rx.upload(
        data,
        with: convertible,
        interceptor: interceptor,
        fileManager: fileManager
    )
}

/**
 Returns an observable of a request using the shared manager instance to upload a specific file to a specified URL.
 The request is started immediately.
 
 - parameter data: An instance of `Data` holding the information of the local file.
 - parameter url: An object adopting `URLConvertible`
 - parameter method: Dio method object
 - parameter headers: A `HTTPHeaders` containing all the additional headers
 - returns: The observable of `UploadRequest` for the created request.
 */
public func upload(
    _ data: Data,
    to convertible: URLConvertible,
    method: HTTPMethod = .post,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    fileManager: FileManager = .default,
    requestModifier: Dio.RequestModifier? = nil
) -> Observable<DioUploadRequest> {
    Dio.default.rx.upload(
        data,
        to: convertible,
        method: method,
        headers: headers,
        interceptor: interceptor,
        fileManager: fileManager,
        requestModifier: requestModifier
    )
}

/**
 Returns an observable of a request progress using the shared manager instance to upload a specific file to a specified URL.
 The request is started immediately.
 
 - parameter data: An instance of `Data` holding the information of the local file.
 - parameter url: An object adopting `URLConvertible`
 - parameter method: Dio method object
 - parameter headers: A `HTTPHeaders` containing all the additional headers
 - returns: The observable of `RxProgress` for the created request.
 */
public func upload(
    _ data: Data,
    to convertible: URLConvertible,
    method: HTTPMethod = .post,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    fileManager: FileManager = .default,
    requestModifier: Dio.RequestModifier? = nil
) -> Observable<RxProgress> {
    Dio.default.rx.upload(
        data,
        to: convertible,
        method: method,
        headers: headers,
        interceptor: interceptor,
        fileManager: fileManager,
        requestModifier: requestModifier
    )
}

/**
 Returns an observable of a request using the shared manager instance to upload any stream to a specified URL.
 The request is started immediately.
 
 - parameter stream: The stream to upload.
 - parameter urlRequest: The request object to start the upload.
 - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
 - returns: The observable of `Request` for the created upload request.
 */
public func upload(
    _ stream: InputStream,
    with convertible: URLRequestConvertible,
    interceptor: RequestInterceptor? = nil,
    fileManager: FileManager = .default
) -> Observable<DioUploadRequest> {
    Dio.default.rx.upload(
        stream,
        with: convertible,
        interceptor: interceptor,
        fileManager: fileManager
    )
}

public func upload(
    _ stream: InputStream,
    with convertible: URLRequestConvertible,
    interceptor: RequestInterceptor? = nil,
    fileManager: FileManager = .default
) -> Observable<RxProgress> {
    Dio.default.rx.upload(
        stream,
        with: convertible,
        interceptor: interceptor,
        fileManager: fileManager
    )
}

/**
 Returns an observable of a request using the shared manager instance to upload a specific file to a specified URL.
 The request is started immediately.
 
 - parameter stream: The `InputStream` to upload.
 - parameter url: An object adopting `URLConvertible`
 - parameter method: Dio method object
 - parameter headers: A `HTTPHeaders` containing all the additional headers
 - returns: The observable of `UploadRequest` for the created request.
 */
public func upload(
    _ stream: InputStream,
    to convertible: URLConvertible,
    method: HTTPMethod = .post,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    fileManager: FileManager = .default,
    requestModifier: Dio.RequestModifier? = nil
) -> Observable<DioUploadRequest> {
    Dio.default.rx.upload(
        stream,
        to: convertible,
        method: method,
        headers: headers,
        interceptor: interceptor,
        fileManager: fileManager,
        requestModifier: requestModifier
    )
}

/**
 Returns an observable of a request progress using the shared manager instance to upload a specific file to a specified URL.
 The request is started immediately.
 
 - parameter stream: The `InputStream` to upload.
 - parameter url: An object adopting `URLConvertible`
 - parameter method: Dio method object
 - parameter headers: A `HTTPHeaders` containing all the additional headers
 - returns: The observable of `RxProgress` for the created request.
 */
public func upload(
    _ stream: InputStream,
    to convertible: URLConvertible,
    method: HTTPMethod = .post,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    fileManager: FileManager = .default,
    requestModifier: Dio.RequestModifier? = nil
) -> Observable<RxProgress> {
    Dio.default.rx.upload(
        stream,
        to: convertible,
        method: method,
        headers: headers,
        interceptor: interceptor,
        fileManager: fileManager,
        requestModifier: requestModifier
    )
}

/**
 Returns an observable of a request using the shared manager instance to upload any stream to a specified URL.
 The request is started immediately.
 
 - parameter multipartFormData: The block for building `MultipartFormData`.
 - parameter urlRequest: The request object to start the upload.
 - returns: The observable of `UploadRequest` for the created upload request.
 */
public func upload(
    multipartFormData: @escaping (MultipartFormData) -> Void,
    with request: URLRequestConvertible,
    usingThreshold encodingMemoryThreshold: UInt64 = MultipartFormData.encodingMemoryThreshold,
    interceptor: RequestInterceptor? = nil,
    fileManager: FileManager = .default
) -> Observable<DioUploadRequest> {
    Dio.default.rx.upload(
        multipartFormData: multipartFormData,
        with: request,
        usingThreshold: encodingMemoryThreshold,
        interceptor: interceptor,
        fileManager: fileManager
    )
}

public func upload(
    multipartFormData: @escaping (MultipartFormData) -> Void,
    with request: URLRequestConvertible,
    usingThreshold encodingMemoryThreshold: UInt64 = MultipartFormData.encodingMemoryThreshold,
    interceptor: RequestInterceptor? = nil,
    fileManager: FileManager = .default
) -> Observable<RxProgress> {
    Dio.default.rx.upload(
        multipartFormData: multipartFormData,
        with: request,
        usingThreshold: encodingMemoryThreshold,
        interceptor: interceptor,
        fileManager: fileManager
    )
}

/**
 Returns an observable of a request using the shared manager instance to upload a specific file to a specified URL.
 The request is started immediately.
 
 - parameter multipartFormData: The block for building `MultipartFormData`.
 - parameter url: An object adopting `URLConvertible`
 - parameter method: Dio method object
 - parameter headers: A `HTTPHeaders` containing all the additional headers
 - returns: The observable of `UploadRequest` for the created request.
 */
public func upload(
    multipartFormData: @escaping (MultipartFormData) -> Void,
    to url: URLConvertible,
    usingThreshold encodingMemoryThreshold: UInt64 = MultipartFormData.encodingMemoryThreshold,
    method: HTTPMethod = .post,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    fileManager: FileManager = .default,
    requestModifier: Dio.RequestModifier? = nil
) -> Observable<DioUploadRequest> {
    Dio.default.rx.upload(
        multipartFormData: multipartFormData,
        to: url,
        usingThreshold: encodingMemoryThreshold,
        method: method,
        headers: headers,
        interceptor: interceptor,
        fileManager: fileManager,
        requestModifier: requestModifier
    )
}

/**
 Returns an observable of a request progress using the shared manager instance to upload a specific file to a specified URL.
 The request is started immediately.
 
 - parameter multipartFormData: The block for building `MultipartFormData`.
 - parameter url: An object adopting `URLConvertible`
 - parameter method: Dio method object
 - parameter headers: A `HTTPHeaders` containing all the additional headers
 - returns: The observable of `RxProgress` for the created request.
 */
public func upload(
    multipartFormData: @escaping (MultipartFormData) -> Void,
    to url: URLConvertible,
    usingThreshold encodingMemoryThreshold: UInt64 = MultipartFormData.encodingMemoryThreshold,
    method: HTTPMethod = .post,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    fileManager: FileManager = .default,
    requestModifier: Dio.RequestModifier? = nil
) -> Observable<RxProgress> {
    Dio.default.rx.upload(
        multipartFormData: multipartFormData,
        to: url,
        usingThreshold: encodingMemoryThreshold,
        method: method,
        headers: headers,
        interceptor: interceptor,
        fileManager: fileManager,
        requestModifier: requestModifier
    )
}

public func upload(
    multipartFormData: MultipartFormData,
    to url: URLConvertible,
    usingThreshold encodingMemoryThreshold: UInt64 = MultipartFormData.encodingMemoryThreshold,
    method: HTTPMethod = .post,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    fileManager: FileManager = .default,
    requestModifier: Dio.RequestModifier? = nil
) -> Observable<DioUploadRequest> {
    Dio.default.rx.upload(
        multipartFormData: multipartFormData,
        to: url,
        usingThreshold: encodingMemoryThreshold,
        method: method,
        headers: headers,
        interceptor: interceptor,
        fileManager: fileManager,
        requestModifier: requestModifier
    )
}

public func upload(
    multipartFormData: MultipartFormData,
    to url: URLConvertible,
    usingThreshold encodingMemoryThreshold: UInt64 = MultipartFormData.encodingMemoryThreshold,
    method: HTTPMethod = .post,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    fileManager: FileManager = .default,
    requestModifier: Dio.RequestModifier? = nil
) -> Observable<RxProgress> {
    Dio.default.rx.upload(
        multipartFormData: multipartFormData,
        to: url,
        usingThreshold: encodingMemoryThreshold,
        method: method,
        headers: headers,
        interceptor: interceptor,
        fileManager: fileManager,
        requestModifier: requestModifier
    )
}

public func upload(
    multipartFormData: MultipartFormData,
    with url: URLRequestConvertible,
    usingThreshold encodingMemoryThreshold: UInt64 = MultipartFormData.encodingMemoryThreshold,
    interceptor: RequestInterceptor? = nil,
    fileManager: FileManager = .default
) -> Observable<DioUploadRequest> {
    Dio.default.rx.upload(
        multipartFormData: multipartFormData,
        with: url,
        usingThreshold: encodingMemoryThreshold,
        interceptor: interceptor,
        fileManager: fileManager
    )
}

public func upload(
    multipartFormData: MultipartFormData,
    with url: URLRequestConvertible,
    usingThreshold encodingMemoryThreshold: UInt64 = MultipartFormData.encodingMemoryThreshold,
    interceptor: RequestInterceptor? = nil,
    fileManager: FileManager = .default
) -> Observable<RxProgress> {
    Dio.default.rx.upload(
        multipartFormData: multipartFormData,
        with: url,
        usingThreshold: encodingMemoryThreshold,
        interceptor: interceptor,
        fileManager: fileManager
    )
}

// MARK: Download

public func download(
    _ convertible: URLConvertible,
    method: HTTPMethod = .get,
    parameters: Parameters? = nil,
    encoding: ParameterEncoding = URLEncoding.default,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    requestModifier: Dio.RequestModifier? = nil,
    to destination: DioDownloadRequest.Destination? = nil
) -> Observable<DioDownloadRequest> {
    Dio.default.rx.download(
        convertible,
        method: method,
        parameters: parameters,
        encoding: encoding,
        headers: headers,
        interceptor: interceptor,
        requestModifier: requestModifier,
        to: destination
    )
}

public func download(
    _ convertible: URLConvertible,
    method: HTTPMethod = .get,
    parameters: Parameters? = nil,
    encoding: ParameterEncoding = URLEncoding.default,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    requestModifier: Dio.RequestModifier? = nil,
    to destination: DioDownloadRequest.Destination? = nil
) -> Observable<RxProgress> {
    Dio.default.rx.download(
        convertible,
        method: method,
        parameters: parameters,
        encoding: encoding,
        headers: headers,
        interceptor: interceptor,
        requestModifier: requestModifier,
        to: destination
    )
}

public func download<Parameters: Encodable>(
    _ convertible: URLConvertible,
    method: HTTPMethod = .get,
    parameters: Parameters? = nil,
    encoder: ParameterEncoder = URLEncodedFormParameterEncoder.default,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    requestModifier: Dio.RequestModifier? = nil,
    to destination: DioDownloadRequest.Destination? = nil
) -> Observable<DioDownloadRequest> {
    Dio.default.rx.download(
        convertible,
        method: method,
        parameters: parameters,
        encoder: encoder,
        headers: headers,
        interceptor: interceptor,
        requestModifier: requestModifier,
        to: destination
    )
}

public func download<Parameters: Encodable>(
    _ convertible: URLConvertible,
    method: HTTPMethod = .get,
    parameters: Parameters? = nil,
    encoder: ParameterEncoder = URLEncodedFormParameterEncoder.default,
    headers: HTTPHeaders? = nil,
    interceptor: RequestInterceptor? = nil,
    requestModifier: Dio.RequestModifier? = nil,
    to destination: DioDownloadRequest.Destination? = nil
) -> Observable<RxProgress> {
    Dio.default.rx.download(
        convertible,
        method: method,
        parameters: parameters,
        encoder: encoder,
        headers: headers,
        interceptor: interceptor,
        requestModifier: requestModifier,
        to: destination
    )
}

/**
 Creates a download request using the shared manager instance for the specified URL request.
 - parameter urlRequest:  The URL request.
 - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
 - parameter destination: The closure used to determine the destination of the downloaded file.
 - returns: The observable of `DownloadRequest` for the created download request.
 */
public func download(
    _ convertible: URLRequestConvertible,
    interceptor: RequestInterceptor? = nil,
    to destination: @escaping DioDownloadRequest.Destination
) -> Observable<DioDownloadRequest> {
    Dio.default.rx.download(
        convertible,
        interceptor: interceptor,
        to: destination
    )
}

public func download(
    _ convertible: URLRequestConvertible,
    interceptor: RequestInterceptor? = nil,
    to destination: @escaping DioDownloadRequest.Destination
) -> Observable<RxProgress> {
    Dio.default.rx.download(
        convertible,
        interceptor: interceptor,
        to: destination
    )
}

// MARK: Resume Data

/**
 Creates a request using the shared manager instance for downloading from the resume data produced from a
 previous request cancellation.
 
 - parameter resumeData:  The resume data. This is an opaque data blob produced by `NSURLSessionDownloadTask`
 when a task is cancelled. See `NSURLSession -downloadTaskWithResumeData:` for additional
 information.
 - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
 - parameter destination: The closure used to determine the destination of the downloaded file.
 - returns: The observable of `Request` for the created download request.
 */
public func download(
    resumingWith: Data,
    interceptor: RequestInterceptor? = nil,
    to destination: @escaping DioDownloadRequest.Destination
) -> Observable<DioDownloadRequest> {
    Dio.default.rx.download(
        resumingWith: resumingWith,
        interceptor: interceptor,
        to: destination
    )
}

public func download(
    resumingWith: Data,
    interceptor: RequestInterceptor? = nil,
    to destination: @escaping DioDownloadRequest.Destination
) -> Observable<RxProgress> {
    Dio.default.rx.download(
        resumingWith: resumingWith,
        interceptor: interceptor,
        to: destination
    )
}

// MARK: Manager - Extension of Manager

extension Dio: ReactiveCompatible {
}

protocol RxDioRequest {
    func responseWith(completionHandler: @escaping (RxDioResponse) -> Void)
    func resume() -> Self
    func cancel() -> Self
}

protocol RxDioResponse {
    var error: Error? { get }
}

extension DioDataResponse: RxDioResponse {
    var error: Error? {
        switch result {
        case let .failure(error):
            return error
        default:
            return nil
        }
    }
}

extension DioDownloadResponse: RxDioResponse {
    var error: Error? {
        switch result {
        case let .failure(error):
            return error
        default:
            return nil
        }
    }
}

extension DioDataRequest: RxDioRequest {
    func responseWith(completionHandler: @escaping (RxDioResponse) -> Void) {
        response { response in
            completionHandler(response)
        }
    }
}

extension DioDownloadRequest: RxDioRequest {
    func responseWith(completionHandler: @escaping (RxDioResponse) -> Void) {
        response { response in
            completionHandler(response)
        }
    }
}

public extension Reactive where Base: Dio {
    // MARK: Generic request convenience
    
    /**
     Creates an observable of the DataRequest.
     
     - parameter createRequest: A function used to create a `Request` using a `Manager`
     
     - returns: A generic observable of created data request
     */
    internal func request<R: RxDioRequest>(_ createRequest: @escaping (Dio) throws -> R) -> Observable<R> {
        Observable.create { observer -> Disposable in
            let request: R
            do {
                request = try createRequest(self.base)
                observer.on(.next(request))
                request.responseWith(completionHandler: { response in
                    if let error = response.error {
                        observer.on(.error(error))
                    } else {
                        observer.on(.completed)
                    }
                })
                
                if !self.base.startRequestsImmediately {
                    _ = request.resume()
                }
                
                return Disposables.create {
                    _ = request.cancel()
                }
            } catch {
                observer.on(.error(error))
                return Disposables.create()
            }
        }
    }
    
    /**
     Creates an observable of the `Request`.
     
     - parameter method: Dio method object
     - parameter url: An object adopting `URLConvertible`
     - parameter parameters: A dictionary containing all necessary options
     - parameter encoding: The kind of encoding used to process parameters
     - parameter header: A dictionary containing all the additional headers
     - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
     - returns: An observable of the `Request`
     */
    func request(
        _ convertible: URLConvertible,
        method: HTTPMethod,
        parameters: Parameters? = nil,
        encoding: ParameterEncoding = URLEncoding.default,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        requestModifier: Dio.RequestModifier? = nil
    ) -> Observable<DioDataRequest> {
        request { manager in
            manager.request(
                convertible,
                method: method,
                parameters: parameters,
                encoding: encoding,
                headers: headers,
                interceptor: interceptor,
                requestModifier: requestModifier
            )
        }
    }
    
    func request<Parameters: Encodable>(
        _ convertible: URLConvertible,
        method: HTTPMethod = .get,
        parameters: Parameters? = nil,
        encoder: ParameterEncoder = URLEncodedFormParameterEncoder.default,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        requestModifier: Dio.RequestModifier? = nil
    ) -> Observable<DioDataRequest> {
        request { manager in
            manager.request(
                convertible,
                method: method,
                parameters: parameters,
                encoder: encoder,
                headers: headers,
                interceptor: interceptor,
                requestModifier: requestModifier
            )
        }
    }
    
    /**
     Creates an observable of the `Request`.
     
     - parameter URLRequest: An object adopting `URLRequestConvertible`
     - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
     - returns: An observable of the `Request`
     */
    func request(
        _ convertible: URLRequestConvertible,
        interceptor: RequestInterceptor? = nil
    ) -> Observable<DioDataRequest> {
        request { manager in
            manager.request(convertible, interceptor: interceptor)
        }
    }
    
    // MARK: response
    
    /**
     Creates an observable of the response
     
     - parameter url: An object adopting `URLConvertible`
     - parameter parameters: A dictionary containing all necessary options
     - parameter encoding: The kind of encoding used to process parameters
     - parameter header: A dictionary containing all the additional headers
     - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
     - returns: An observable of  `NSHTTPURLResponse`
     */
    func response(
        _ convertible: URLConvertible,
        method: HTTPMethod,
        parameters: Parameters? = nil,
        encoding: ParameterEncoding = URLEncoding.default,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        requestModifier: Dio.RequestModifier? = nil
    ) -> Observable<HTTPURLResponse> {
        request(
            convertible,
            method: method,
            parameters: parameters,
            encoding: encoding,
            headers: headers,
            interceptor: interceptor,
            requestModifier: requestModifier
        )
        .flatMap {
            $0.rx.response()
        }
    }
    
    // MARK: data
    
    /**
     Creates an observable of the data.
     
     - parameter url: An object adopting `URLConvertible`
     - parameter parameters: A dictionary containing all necessary options
     - parameter encoding: The kind of encoding used to process parameters
     - parameter header: A dictionary containing all the additional headers
     - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
     - returns: An observable of the tuple `(NSHTTPURLResponse, NSData)`
     */
    func responseData(
        _ convertible: URLConvertible,
        method: HTTPMethod,
        parameters: Parameters? = nil,
        encoding: ParameterEncoding = URLEncoding.default,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        requestModifier: Dio.RequestModifier? = nil
    ) -> Observable<(HTTPURLResponse, Data)> {
        request(
            convertible,
            method: method,
            parameters: parameters,
            encoding: encoding,
            headers: headers,
            interceptor: interceptor,
            requestModifier: requestModifier
        )
        .flatMap {
            $0.rx.responseData()
        }
    }
    
    /**
     Creates an observable of the data.
     
     - parameter URLRequest: An object adopting `URLRequestConvertible`
     - parameter parameters: A dictionary containing all necessary options
     - parameter encoding: The kind of encoding used to process parameters
     - parameter header: A dictionary containing all the additional headers
     - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
     - returns: An observable of `NSData`
     */
    func data(
        _ convertible: URLConvertible,
        method: HTTPMethod,
        parameters: Parameters? = nil,
        encoding: ParameterEncoding = URLEncoding.default,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        requestModifier: Dio.RequestModifier? = nil
    ) -> Observable<Data> {
        request(
            convertible,
            method: method,
            parameters: parameters,
            encoding: encoding,
            headers: headers,
            interceptor: interceptor,
            requestModifier: requestModifier
        )
        .flatMap {
            $0.rx.data()
        }
    }
    
    // MARK: string
    
    /**
     Creates an observable of the tuple `(NSHTTPURLResponse, String)`.
     
     - parameter url: An object adopting `URLRequestConvertible`
     - parameter parameters: A dictionary containing all necessary options
     - parameter encoding: The kind of encoding used to process parameters
     - parameter header: A dictionary containing all the additional headers
     - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
     - returns: An observable of the tuple `(NSHTTPURLResponse, String)`
     */
    func responseString(
        _ convertible: URLConvertible,
        method: HTTPMethod,
        parameters: Parameters? = nil,
        encoding: ParameterEncoding = URLEncoding.default,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        requestModifier: Dio.RequestModifier? = nil
    ) -> Observable<(HTTPURLResponse, String)> {
        request(
            convertible,
            method: method,
            parameters: parameters,
            encoding: encoding,
            headers: headers,
            interceptor: interceptor,
            requestModifier: requestModifier
        )
        .flatMap {
            $0.rx.responseString()
        }
    }
    
    /**
     Creates an observable of the data encoded as String.
     
     - parameter url: An object adopting `URLConvertible`
     - parameter parameters: A dictionary containing all necessary options
     - parameter encoding: The kind of encoding used to process parameters
     - parameter header: A dictionary containing all the additional headers
     - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
     - returns: An observable of `String`
     */
    func string(
        _ convertible: URLConvertible,
        method: HTTPMethod,
        parameters: Parameters? = nil,
        encoding: ParameterEncoding = URLEncoding.default,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        requestModifier: Dio.RequestModifier? = nil
    ) -> Observable<String> {
        request(
            convertible,
            method: method,
            parameters: parameters,
            encoding: encoding,
            headers: headers,
            interceptor: interceptor,
            requestModifier: requestModifier
        )
        .flatMap { (request) -> Observable<String> in
            request.rx.string()
        }
    }
    
    // MARK: JSON
    
    /**
     Creates an observable of the data decoded from JSON and processed as tuple `(NSHTTPURLResponse, AnyObject)`.
     
     - parameter url: An object adopting `URLRequestConvertible`
     - parameter parameters: A dictionary containing all necessary options
     - parameter encoding: The kind of encoding used to process parameters
     - parameter header: A dictionary containing all the additional headers
     - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
     - returns: An observable of the tuple `(NSHTTPURLResponse, AnyObject)`
     */
    @available(*, deprecated, message: "responseJSON deprecated and will be removed in Dio 6. Use responseDecodable instead.")
    func responseJSON(
        _ convertible: URLConvertible,
        method: HTTPMethod,
        parameters: Parameters? = nil,
        encoding: ParameterEncoding = URLEncoding.default,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        requestModifier: Dio.RequestModifier? = nil
    ) -> Observable<(HTTPURLResponse, Any)> {
        request(
            convertible,
            method: method,
            parameters: parameters,
            encoding: encoding,
            headers: headers,
            interceptor: interceptor,
            requestModifier: requestModifier
        )
        .flatMap {
            $0.rx.responseJSON()
        }
    }
    
    /**
     Creates an observable of the data decoded from JSON and processed as `AnyObject`.
     
     - parameter URLRequest: An object adopting `URLRequestConvertible`
     - parameter parameters: A dictionary containing all necessary options
     - parameter encoding: The kind of encoding used to process parameters
     - parameter header: A dictionary containing all the additional headers
     - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
     - returns: An observable of `AnyObject`
     */
    @available(*, deprecated, message: "json deprecated and will be removed in Dio 6. Use decodable instead.")
    func json(
        _ convertible: URLConvertible,
        method: HTTPMethod,
        parameters: Parameters? = nil,
        encoding: ParameterEncoding = URLEncoding.default,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        requestModifier: Dio.RequestModifier? = nil
    ) -> Observable<Any> {
        request(
            convertible,
            method: method,
            parameters: parameters,
            encoding: encoding,
            headers: headers,
            interceptor: interceptor,
            requestModifier: requestModifier
        )
        .flatMap {
            $0.rx.json()
        }
    }
    
    // MARK: Decodable
    
    /**
     Creates an observable of the data decoded from Decodable and processed as tuple `(NSHTTPURLResponse, T)`.
     
     - parameter url: An object adopting `URLRequestConvertible`
     - parameter parameters: A dictionary containing all necessary options
     - parameter encoding: The kind of encoding used to process parameters
     - parameter header: A dictionary containing all the additional headers
     - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
     - returns: An observable of the tuple `(NSHTTPURLResponse, T)`
     */
    func responseDecodable<T: Decodable>(
        _ convertible: URLConvertible,
        of type: T.Type = T.self,
        method: HTTPMethod,
        parameters: Parameters? = nil,
        encoding: ParameterEncoding = URLEncoding.default,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        requestModifier: Dio.RequestModifier? = nil
    ) -> Observable<(HTTPURLResponse, T)> {
        request(
            convertible,
            method: method,
            parameters: parameters,
            encoding: encoding,
            headers: headers,
            interceptor: interceptor,
            requestModifier: requestModifier
        )
        .flatMap {
            $0.rx.responseDecodable(of: type)
        }
    }
    
    /**
     Creates an observable of the data decoded from Decodable and processed as `T`.
     
     - parameter URLRequest: An object adopting `URLRequestConvertible`
     - parameter parameters: A dictionary containing all necessary options
     - parameter encoding: The kind of encoding used to process parameters
     - parameter header: A dictionary containing all the additional headers
     - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
     - returns: An observable of `T`
     */
    func decodable<T: Decodable>(
        _ convertible: URLConvertible,
        of type: T.Type = T.self,
        method: HTTPMethod,
        parameters: Parameters? = nil,
        encoding: ParameterEncoding = URLEncoding.default,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        requestModifier: Dio.RequestModifier? = nil
    ) -> Observable<T> {
        request(
            convertible,
            method: method,
            parameters: parameters,
            encoding: encoding,
            headers: headers,
            interceptor: interceptor,
            requestModifier: requestModifier
        )
        .flatMap {
            $0.rx.decodable(of: type)
        }
    }
    
    // MARK: Upload
    
    /**
     Returns an observable of a request using the shared manager instance to upload a specific file to a specified URL.
     The request is started immediately.
     
     - parameter file: An instance of NSURL holding the information of the local file.
     - parameter urlRequest: The request object to start the upload.
     - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
     - returns: The observable of `AnyObject` for the created request.
     */
    func upload(
        _ file: URL,
        with convertible: URLRequestConvertible,
        interceptor: RequestInterceptor? = nil,
        fileManager: FileManager = .default
    ) -> Observable<DioUploadRequest> {
        request { manager in
            manager.upload(
                file,
                with: convertible,
                interceptor: interceptor,
                fileManager: fileManager
            )
        }
    }
    
    
    func upload(
        _ file: URL,
        with convertible: URLRequestConvertible,
        interceptor: RequestInterceptor? = nil,
        fileManager: FileManager = .default
    ) -> Observable<RxProgress> {
        upload(
            file,
            with: convertible,
            interceptor: interceptor,
            fileManager: fileManager
        )
        .flatMap {
            $0.rx.progress()
        }
    }
    
    /**
     Returns an observable of a request using the shared manager instance to upload a specific file to a specified URL.
     The request is started immediately.
     
     - parameter file: An instance of `URL` holding the information of the local file.
     - parameter url: An object adopting `URLConvertible`
     - parameter method: Dio method object
     - parameter headers: A `HTTPHeaders` containing all the additional headers
     - returns: The observable of `UploadRequest` for the created request.
     */
    func upload(
        _ file: URL,
        to convertible: URLConvertible,
        method: HTTPMethod = .post,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        fileManager: FileManager = .default,
        requestModifier: Dio.RequestModifier? = nil
    ) -> Observable<DioUploadRequest> {
        request { manager in
            manager.upload(
                file,
                to: convertible,
                method: method,
                headers: headers,
                interceptor: interceptor,
                fileManager: fileManager,
                requestModifier: requestModifier
            )
        }
    }
    
    /**
     Returns an observable of a request progress using the shared manager instance to upload a specific file to a specified URL.
     The request is started immediately.
     
     - parameter file: An instance of `URL` holding the information of the local file.
     - parameter url: An object adopting `URLConvertible`
     - parameter method: Dio method object
     - parameter headers: A `HTTPHeaders` containing all the additional headers
     - returns: The observable of `RxProgress` for the created request.
     */
    func upload(
        _ file: URL,
        to convertible: URLConvertible,
        method: HTTPMethod = .post,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        fileManager: FileManager = .default,
        requestModifier: Dio.RequestModifier? = nil
    ) -> Observable<RxProgress> {
        upload(
            file,
            to: convertible,
            method: method,
            headers: headers,
            interceptor: interceptor,
            fileManager: fileManager,
            requestModifier: requestModifier
        )
        .flatMap {
            $0.rx.progress()
        }
    }
    
    /**
     Returns an observable of a request using the shared manager instance to upload any data to a specified URL.
     The request is started immediately.
     
     - parameter data: An instance of Data holdint the data to upload.
     - parameter urlRequest: The request object to start the upload.
     - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
     - returns: The observable of `UploadRequest` for the created request.
     */
    func upload(
        _ data: Data,
        with convertible: URLRequestConvertible,
        interceptor: RequestInterceptor? = nil,
        fileManager: FileManager = .default
    ) -> Observable<DioUploadRequest> {
        request { manager in
            manager.upload(
                data,
                with: convertible,
                interceptor: interceptor,
                fileManager: fileManager
            )
        }
    }
    
    func upload(
        _ data: Data,
        with convertible: URLRequestConvertible,
        interceptor: RequestInterceptor? = nil,
        fileManager: FileManager = .default
    ) -> Observable<RxProgress> {
        upload(
            data,
            with: convertible,
            interceptor: interceptor,
            fileManager: fileManager
        )
        .flatMap {
            $0.rx.progress()
        }
    }
    
    /**
     Returns an observable of a request using the shared manager instance to upload a specific file to a specified URL.
     The request is started immediately.
     
     - parameter data: An instance of `Data` holding the information of the local file.
     - parameter url: An object adopting `URLConvertible`
     - parameter method: Dio method object
     - parameter headers: A `HTTPHeaders` containing all the additional headers
     - returns: The observable of `UploadRequest` for the created request.
     */
    func upload(
        _ data: Data,
        to convertible: URLConvertible,
        method: HTTPMethod = .post,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        fileManager: FileManager = .default,
        requestModifier: Dio.RequestModifier? = nil
    ) -> Observable<DioUploadRequest> {
        request { manager in
            manager.upload(
                data,
                to: convertible,
                method: method,
                headers: headers,
                interceptor: interceptor,
                fileManager: fileManager,
                requestModifier: requestModifier
            )
        }
    }
    
    /**
     Returns an observable of a request progress using the shared manager instance to upload a specific file to a specified URL.
     The request is started immediately.
     
     - parameter data: An instance of `Data` holding the information of the local file.
     - parameter url: An object adopting `URLConvertible`
     - parameter method: Dio method object
     - parameter headers: A `HTTPHeaders` containing all the additional headers
     - returns: The observable of `RxProgress` for the created request.
     */
    func upload(
        _ data: Data,
        to convertible: URLConvertible,
        method: HTTPMethod = .post,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        fileManager: FileManager = .default,
        requestModifier: Dio.RequestModifier? = nil
    ) -> Observable<RxProgress> {
        upload(
            data,
            to: convertible,
            method: method,
            headers: headers,
            interceptor: interceptor,
            fileManager: fileManager,
            requestModifier: requestModifier
        )
        .flatMap {
            $0.rx.progress()
        }
    }
    
    /**
     Returns an observable of a request using the shared manager instance to upload any stream to a specified URL.
     The request is started immediately.
     
     - parameter stream: The stream to upload.
     - parameter urlRequest: The request object to start the upload.
     - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
     - returns: The observable of `(NSData?, RxProgress)` for the created upload request.
     */
    func upload(
        _ stream: InputStream,
        with convertible: URLRequestConvertible,
        interceptor: RequestInterceptor? = nil,
        fileManager: FileManager = .default
    ) -> Observable<DioUploadRequest> {
        request { manager in
            manager.upload(
                stream,
                with: convertible,
                interceptor: interceptor,
                fileManager: fileManager
            )
        }
    }
    
    func upload(
        _ stream: InputStream,
        with convertible: URLRequestConvertible,
        interceptor: RequestInterceptor? = nil,
        fileManager: FileManager = .default
    ) -> Observable<RxProgress> {
        upload(
            stream,
            with: convertible,
            interceptor: interceptor,
            fileManager: fileManager
        )
        .flatMap {
            $0.rx.progress()
        }
    }
    
    /**
     Returns an observable of a request using the shared manager instance to upload a specific file to a specified URL.
     The request is started immediately.
     
     - parameter stream: The `InputStream` to upload.
     - parameter url: An object adopting `URLConvertible`
     - parameter method: Dio method object
     - parameter headers: A `HTTPHeaders` containing all the additional headers
     - returns: The observable of `UploadRequest` for the created request.
     */
    func upload(
        _ stream: InputStream,
        to convertible: URLConvertible,
        method: HTTPMethod = .post,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        fileManager: FileManager = .default,
        requestModifier: Dio.RequestModifier? = nil
    ) -> Observable<DioUploadRequest> {
        request { manager in
            manager.upload(
                stream,
                to: convertible,
                method: method,
                headers: headers,
                interceptor: interceptor,
                fileManager: fileManager,
                requestModifier: requestModifier
            )
        }
    }
    
    /**
     Returns an observable of a request progress using the shared manager instance to upload a specific file to a specified URL.
     The request is started immediately.
     
     - parameter stream: The `InputStream` to upload.
     - parameter url: An object adopting `URLConvertible`
     - parameter method: Dio method object
     - parameter headers: A `HTTPHeaders` containing all the additional headers
     - returns: The observable of `RxProgress` for the created request.
     */
    func upload(
        _ stream: InputStream,
        to convertible: URLConvertible,
        method: HTTPMethod = .post,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        fileManager: FileManager = .default,
        requestModifier: Dio.RequestModifier? = nil
    ) -> Observable<RxProgress> {
        upload(stream, to: convertible, method: method, headers: headers)
            .flatMap {
                $0.rx.progress()
            }
    }
    
    /**
     Returns an observable of a request using the shared manager instance to upload any stream to a specified URL.
     The request is started immediately.
     
     - parameter multipartFormData: The block for building `MultipartFormData`.
     - parameter urlRequest: The request object to start the upload.
     - returns: The observable of `UploadRequest` for the created upload request.
     */
    func upload(
        multipartFormData: @escaping (MultipartFormData) -> Void,
        with url: URLRequestConvertible,
        usingThreshold encodingMemoryThreshold: UInt64 = MultipartFormData.encodingMemoryThreshold,
        interceptor: RequestInterceptor? = nil,
        fileManager: FileManager = .default
    ) -> Observable<DioUploadRequest> {
        request { manager in
            manager.upload(
                multipartFormData: multipartFormData,
                with: url,
                usingThreshold: encodingMemoryThreshold,
                interceptor: interceptor,
                fileManager: fileManager
            )
        }
    }
    
    func upload(
        multipartFormData: @escaping (MultipartFormData) -> Void,
        with url: URLRequestConvertible,
        usingThreshold encodingMemoryThreshold: UInt64 = MultipartFormData.encodingMemoryThreshold,
        interceptor: RequestInterceptor? = nil,
        fileManager: FileManager = .default
    ) -> Observable<RxProgress> {
        upload(
            multipartFormData: multipartFormData,
            with: url,
            usingThreshold: encodingMemoryThreshold,
            interceptor: interceptor,
            fileManager: fileManager
        )
        .flatMap {
            $0.rx.progress()
        }
    }
    
    func upload(
        multipartFormData: MultipartFormData,
        to url: URLConvertible,
        usingThreshold encodingMemoryThreshold: UInt64 = MultipartFormData.encodingMemoryThreshold,
        method: HTTPMethod = .post,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        fileManager: FileManager = .default,
        requestModifier: Dio.RequestModifier? = nil
    ) -> Observable<DioUploadRequest> {
        request { manager in
            manager.upload(
                multipartFormData: multipartFormData,
                to: url,
                usingThreshold: encodingMemoryThreshold,
                method: method,
                headers: headers,
                interceptor: interceptor,
                fileManager: fileManager,
                requestModifier: requestModifier
            )
        }
    }
    
    func upload(
        multipartFormData: MultipartFormData,
        to url: URLConvertible,
        usingThreshold encodingMemoryThreshold: UInt64 = MultipartFormData.encodingMemoryThreshold,
        method: HTTPMethod = .post,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        fileManager: FileManager = .default,
        requestModifier: Dio.RequestModifier? = nil
    ) -> Observable<RxProgress> {
        upload(
            multipartFormData: multipartFormData,
            to: url,
            usingThreshold: encodingMemoryThreshold,
            method: method,
            headers: headers,
            interceptor: interceptor,
            fileManager: fileManager,
            requestModifier: requestModifier
        )
        .flatMap {
            $0.rx.progress()
        }
    }
    
    func upload(
        multipartFormData: MultipartFormData,
        with url: URLRequestConvertible,
        usingThreshold encodingMemoryThreshold: UInt64 = MultipartFormData.encodingMemoryThreshold,
        interceptor: RequestInterceptor? = nil,
        fileManager: FileManager = .default
    ) -> Observable<DioUploadRequest> {
        request { manager in
            manager.upload(
                multipartFormData: multipartFormData,
                with: url,
                usingThreshold: encodingMemoryThreshold,
                interceptor: interceptor,
                fileManager: fileManager
            )
        }
    }
    
    func upload(
        multipartFormData: MultipartFormData,
        with url: URLRequestConvertible,
        usingThreshold encodingMemoryThreshold: UInt64 = MultipartFormData.encodingMemoryThreshold,
        interceptor: RequestInterceptor? = nil,
        fileManager: FileManager = .default
    ) -> Observable<RxProgress> {
        upload(
            multipartFormData: multipartFormData,
            with: url,
            usingThreshold: encodingMemoryThreshold,
            interceptor: interceptor,
            fileManager: fileManager
        )
        .flatMap {
            $0.rx.progress()
        }
    }
    
    /**
     Returns an observable of a request using the shared manager instance to upload a specific file to a specified URL.
     The request is started immediately.
     
     - parameter multipartFormData: The block for building `MultipartFormData`.
     - parameter url: An object adopting `URLConvertible`
     - parameter method: Dio method object
     - parameter headers: A `HTTPHeaders` containing all the additional headers
     - returns: The observable of `UploadRequest` for the created request.
     */
    func upload(
        multipartFormData: @escaping (MultipartFormData) -> Void,
        to url: URLConvertible,
        usingThreshold encodingMemoryThreshold: UInt64 = MultipartFormData.encodingMemoryThreshold,
        method: HTTPMethod = .post,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        fileManager: FileManager = .default,
        requestModifier: Dio.RequestModifier? = nil
    ) -> Observable<DioUploadRequest> {
        request { manager in
            manager.upload(
                multipartFormData: multipartFormData,
                to: url,
                usingThreshold: encodingMemoryThreshold,
                method: method,
                headers: headers,
                interceptor: interceptor,
                fileManager: fileManager,
                requestModifier: requestModifier
            )
        }
    }
    
    /**
     Returns an observable of a request progress using the shared manager instance to upload a specific file to a specified URL.
     The request is started immediately.
     
     - parameter multipartFormData: The block for building `MultipartFormData`.
     - parameter url: An object adopting `URLConvertible`
     - parameter method: Dio method object
     - parameter headers: A `HTTPHeaders` containing all the additional headers
     - returns: The observable of `RxProgress` for the created request.
     */
    func upload(
        multipartFormData: @escaping (MultipartFormData) -> Void,
        to url: URLConvertible,
        usingThreshold encodingMemoryThreshold: UInt64 = MultipartFormData.encodingMemoryThreshold,
        method: HTTPMethod = .post,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        fileManager: FileManager = .default,
        requestModifier: Dio.RequestModifier? = nil
    ) -> Observable<RxProgress> {
        upload(
            multipartFormData: multipartFormData,
            to: url,
            usingThreshold: encodingMemoryThreshold,
            method: method,
            headers: headers,
            interceptor: interceptor,
            fileManager: fileManager,
            requestModifier: requestModifier
        )
        .flatMap {
            $0.rx.progress()
        }
    }
    
    // MARK: Download
    
    func download(
        _ convertible: URLConvertible,
        method: HTTPMethod = .get,
        parameters: Parameters? = nil,
        encoding: ParameterEncoding = URLEncoding.default,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        requestModifier: Dio.RequestModifier? = nil,
        to destination: DioDownloadRequest.Destination? = nil
    ) -> Observable<DioDownloadRequest> {
        request { manager in
            manager.download(
                convertible,
                method: method,
                parameters: parameters,
                encoding: encoding,
                headers: headers,
                interceptor: interceptor,
                requestModifier: requestModifier,
                to: destination
            )
        }
    }
    
    func download(
        _ convertible: URLConvertible,
        method: HTTPMethod = .get,
        parameters: Parameters? = nil,
        encoding: ParameterEncoding = URLEncoding.default,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        requestModifier: Dio.RequestModifier? = nil,
        to destination: DioDownloadRequest.Destination? = nil
    ) -> Observable<RxProgress> {
        download(
            convertible,
            method: method,
            parameters: parameters,
            encoding: encoding,
            headers: headers,
            interceptor: interceptor,
            requestModifier: requestModifier,
            to: destination
        )
        .flatMap {
            $0.rx.progress()
        }
    }
    
    func download<Parameters: Encodable>(
        _ convertible: URLConvertible,
        method: HTTPMethod = .get,
        parameters: Parameters? = nil,
        encoder: ParameterEncoder = URLEncodedFormParameterEncoder.default,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        requestModifier: Dio.RequestModifier? = nil,
        to destination: DioDownloadRequest.Destination? = nil
    ) -> Observable<DioDownloadRequest> {
        request { manager in
            manager.download(
                convertible,
                method: method,
                parameters: parameters,
                encoder: encoder,
                headers: headers,
                interceptor: interceptor,
                requestModifier: requestModifier,
                to: destination
            )
        }
    }
    
    func download<Parameters: Encodable>(
        _ convertible: URLConvertible,
        method: HTTPMethod = .get,
        parameters: Parameters? = nil,
        encoder: ParameterEncoder = URLEncodedFormParameterEncoder.default,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        requestModifier: Dio.RequestModifier? = nil,
        to destination: DioDownloadRequest.Destination? = nil
    ) -> Observable<RxProgress> {
        download(
            convertible,
            method: method,
            parameters: parameters,
            encoder: encoder,
            headers: headers,
            interceptor: interceptor,
            requestModifier: requestModifier,
            to: destination
        )
        .flatMap {
            $0.rx.progress()
        }
    }
    
    /**
     Creates a download request using the shared manager instance for the specified URL request.
     - parameter urlRequest:  The URL request.
     - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
     - parameter destination: The closure used to determine the destination of the downloaded file.
     - returns: The observable of `(NSData?, RxProgress)` for the created download request.
     */
    func download(
        _ convertible: URLRequestConvertible,
        interceptor: RequestInterceptor? = nil,
        to destination: @escaping DioDownloadRequest.Destination
    ) -> Observable<DioDownloadRequest> {
        request { manager in
            manager.download(
                convertible,
                interceptor: interceptor,
                to: destination
            )
        }
    }
    
    func download(
        _ convertible: URLRequestConvertible,
        interceptor: RequestInterceptor? = nil,
        to destination: @escaping DioDownloadRequest.Destination
    ) -> Observable<RxProgress> {
        download(
            convertible,
            interceptor: interceptor,
            to: destination
        )
        .flatMap {
            $0.rx.progress()
        }
    }
    
    /**
     Creates a request using the shared manager instance for downloading with a resume data produced from a
     previous request cancellation.
     
     - parameter resumeData:  The resume data. This is an opaque data blob produced by `NSURLSessionDownloadTask`
     when a task is cancelled. See `NSURLSession -downloadTaskWithResumeData:` for additional
     information.
     - parameter interceptor: `RequestInterceptor` value to be used by the returned `DataRequest`. `nil` by default.
     - parameter destination: The closure used to determine the destination of the downloaded file.
     - returns: The observable of `(NSData?, RxProgress)` for the created download request.
     */
    func download(
        resumingWith: Data,
        interceptor: RequestInterceptor? = nil,
        to destination: @escaping DioDownloadRequest.Destination
    ) -> Observable<DioDownloadRequest> {
        request { manager in
            manager.download(
                resumingWith: resumingWith,
                interceptor: interceptor,
                to: destination
            )
        }
    }
    
    func download(
        resumingWith: Data,
        interceptor: RequestInterceptor? = nil,
        to destination: @escaping DioDownloadRequest.Destination
    ) -> Observable<RxProgress> {
        download(
            resumingWith: resumingWith,
            interceptor: interceptor,
            to: destination
        )
        .flatMap {
            $0.rx.progress()
        }
    }
}

#if swift(>=5.5)
public extension Reactive where Base: Dio {
    // MARK: Generic request convenience
    
    func request(
        _ convertible: URLConvertible,
        method: HTTPMethod,
        parameters: Parameters? = nil,
        encoding: ParameterEncoding = URLEncoding.default,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        requestModifier: Dio.RequestModifier? = nil
    ) -> DioDataRequest {
        base.request(
            convertible,
            method: method,
            parameters: parameters,
            encoding: encoding,
            headers: headers,
            interceptor: interceptor,
            requestModifier: requestModifier
        )
    }
    
    func request<Parameters: Encodable>(
        _ convertible: URLConvertible,
        method: HTTPMethod = .get,
        parameters: Parameters? = nil,
        encoder: ParameterEncoder = URLEncodedFormParameterEncoder.default,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        requestModifier: Dio.RequestModifier? = nil
    ) -> DioDataRequest {
        base.request(
            convertible,
            method: method,
            parameters: parameters,
            encoder: encoder,
            headers: headers,
            interceptor: interceptor,
            requestModifier: requestModifier
        )
    }
    
    func request(
        _ convertible: URLRequestConvertible,
        interceptor: RequestInterceptor? = nil
    ) -> DioDataRequest {
        base.request(convertible, interceptor: interceptor)
    }
    
    // MARK: response
    
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    func response(
        _ convertible: URLConvertible,
        method: HTTPMethod,
        parameters: Parameters? = nil,
        encoding: ParameterEncoding = URLEncoding.default,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        requestModifier: Dio.RequestModifier? = nil
    ) async throws -> HTTPURLResponse {
        try await request(
            convertible,
            method: method,
            parameters: parameters,
            encoding: encoding,
            headers: headers,
            interceptor: interceptor,
            requestModifier: requestModifier
        )
        .rx.response()
    }
    
    // MARK: data
    
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    func responseData(
        _ convertible: URLConvertible,
        method: HTTPMethod,
        parameters: Parameters? = nil,
        encoding: ParameterEncoding = URLEncoding.default,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        requestModifier: Dio.RequestModifier? = nil
    ) async throws -> (HTTPURLResponse, Data) {
        try await request(
            convertible,
            method: method,
            parameters: parameters,
            encoding: encoding,
            headers: headers,
            interceptor: interceptor,
            requestModifier: requestModifier
        )
        .rx.responseData()
    }
    
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    func data(
        _ convertible: URLConvertible,
        method: HTTPMethod,
        parameters: Parameters? = nil,
        encoding: ParameterEncoding = URLEncoding.default,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        requestModifier: Dio.RequestModifier? = nil
    ) async throws -> Data {
        try await request(
            convertible,
            method: method,
            parameters: parameters,
            encoding: encoding,
            headers: headers,
            interceptor: interceptor,
            requestModifier: requestModifier
        )
        .rx.data()
    }
    
    // MARK: string
    
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    func responseString(
        _ convertible: URLConvertible,
        method: HTTPMethod,
        parameters: Parameters? = nil,
        encoding: ParameterEncoding = URLEncoding.default,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        requestModifier: Dio.RequestModifier? = nil
    ) async throws -> (HTTPURLResponse, String) {
        try await request(
            convertible,
            method: method,
            parameters: parameters,
            encoding: encoding,
            headers: headers,
            interceptor: interceptor,
            requestModifier: requestModifier
        )
        .rx.responseString()
    }
    
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    func string(
        _ convertible: URLConvertible,
        method: HTTPMethod,
        parameters: Parameters? = nil,
        encoding: ParameterEncoding = URLEncoding.default,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        requestModifier: Dio.RequestModifier? = nil
    ) async throws -> String {
        try await  request(
            convertible,
            method: method,
            parameters: parameters,
            encoding: encoding,
            headers: headers,
            interceptor: interceptor,
            requestModifier: requestModifier
        )
        .rx.string()
    }
    
    // MARK: JSON
    
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    @available(*, deprecated, message: "responseJSON deprecated and will be removed in Dio 6. Use responseDecodable instead.")
    func responseJSON(
        _ convertible: URLConvertible,
        method: HTTPMethod,
        parameters: Parameters? = nil,
        encoding: ParameterEncoding = URLEncoding.default,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        requestModifier: Dio.RequestModifier? = nil
    ) async throws -> (HTTPURLResponse, Any) {
        try await request(
            convertible,
            method: method,
            parameters: parameters,
            encoding: encoding,
            headers: headers,
            interceptor: interceptor,
            requestModifier: requestModifier
        )
        .rx.responseJSON()
    }
    
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    @available(*, deprecated, message: "json deprecated and will be removed in Dio 6. Use decodable instead.")
    func json(
        _ convertible: URLConvertible,
        method: HTTPMethod,
        parameters: Parameters? = nil,
        encoding: ParameterEncoding = URLEncoding.default,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        requestModifier: Dio.RequestModifier? = nil
    ) async throws -> Any {
        try await request(
            convertible,
            method: method,
            parameters: parameters,
            encoding: encoding,
            headers: headers,
            interceptor: interceptor,
            requestModifier: requestModifier
        )
        .rx.json()
    }
    
    // MARK: Decodable
    
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    func responseDecodable<T: Decodable>(
        _ convertible: URLConvertible,
        of type: T.Type = T.self,
        method: HTTPMethod,
        parameters: Parameters? = nil,
        encoding: ParameterEncoding = URLEncoding.default,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        requestModifier: Dio.RequestModifier? = nil
    ) async throws -> (HTTPURLResponse, T) {
        try await request(
            convertible,
            method: method,
            parameters: parameters,
            encoding: encoding,
            headers: headers,
            interceptor: interceptor,
            requestModifier: requestModifier
        )
        .rx.responseDecodable(of: type)
    }
    
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    func decodable<T: Decodable>(
        _ convertible: URLConvertible,
        of type: T.Type = T.self,
        method: HTTPMethod,
        parameters: Parameters? = nil,
        encoding: ParameterEncoding = URLEncoding.default,
        headers: HTTPHeaders? = nil,
        interceptor: RequestInterceptor? = nil,
        requestModifier: Dio.RequestModifier? = nil
    ) async throws -> T {
        try await request(
            convertible,
            method: method,
            parameters: parameters,
            encoding: encoding,
            headers: headers,
            interceptor: interceptor,
            requestModifier: requestModifier
        )
        .rx.decodable(of: type)
    }
}
#endif

// MARK: Request - Common Response Handlers

public extension ObservableType where Element == DioDataRequest {
    @available(*, deprecated, message: "responseJSON deprecated and will be removed in Dio 6. ")
    func responseJSON() -> Observable<DioDataResponse<Any>> {
        flatMap {
            $0.rx.responseJSON()
        }
    }
    
    @available(*, deprecated, message: "json deprecated and will be removed in Dio 6. ")
    func json(options: JSONSerialization.ReadingOptions = .allowFragments) -> Observable<Any> {
        flatMap {
            $0.rx.json(options: options)
        }
    }
    
    func responseString(encoding: String.Encoding? = nil) -> Observable<(HTTPURLResponse, String)> {
        flatMap {
            $0.rx.responseString(encoding: encoding)
        }
    }
    
    func string(encoding: String.Encoding? = nil) -> Observable<String> {
        flatMap {
            $0.rx.string(encoding: encoding)
        }
    }
    
    func responseData() -> Observable<(HTTPURLResponse, Data)> {
        flatMap {
            $0.rx.responseData()
        }
    }
    
    func response() -> Observable<HTTPURLResponse> {
        flatMap {
            $0.rx.response()
        }
    }
    
    func data() -> Observable<Data> {
        flatMap {
            $0.rx.data()
        }
    }
    
    func progress() -> Observable<RxProgress> {
        flatMap {
            $0.rx.progress()
        }
    }
}

// MARK: Request - Validation

public extension ObservableType where Element == DioDataRequest {
    func validate<S: Sequence>(statusCode: S) -> Observable<Element> where S.Element == Int {
        map {
            $0.validate(statusCode: statusCode)
        }
    }
    
    func validate() -> Observable<Element> {
        map {
            $0.validate()
        }
    }
    
    func validate<S: Sequence>(contentType acceptableContentTypes: S) -> Observable<Element> where S.Iterator.Element == String {
        map {
            $0.validate(contentType: acceptableContentTypes)
        }
    }
    
    func validate(_ validation: @escaping DioDataRequest.Validation) -> Observable<Element> {
        map {
            $0.validate(validation)
        }
    }
}

#if swift(>=5.5)
public extension ObservableType where Element == DioDataRequest {
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    func asFuture() async throws -> Element {
        let continuation = CheckedCancellationThrowingContinuation({ (continuation: CheckedContinuation<Element, Error>) -> Disposable in
            self.subscribe(onNext: { request in
                continuation.resume(returning: request)
            }, onError: { error in
                continuation.resume(throwing: error)
            })
        }, onCancel: { (disposable: Disposable) in
            disposable.dispose()
        })
        return try await continuation.with()
    }
}

@available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
class CheckedCancellationThrowingContinuation<T, R> {
    private var result: R? = nil
    private let function: String
    private let body: (_Concurrency.CheckedContinuation<T, Error>) -> R
    private let handler: (R) -> Void
    
    init(function: String = #function, _ body: @escaping (CheckedContinuation<T, Error>) -> R, onCancel handler: @escaping (R) -> ()) {
        self.function = function
        self.body = body
        self.handler = handler
    }
    
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    func with() async throws -> T {
        try await withTaskCancellationHandler {
            try await withCheckedThrowingContinuation(function: function) { (continuation: CheckedContinuation<T, Error>) in
                result = body(continuation)
            }
        } onCancel: {
            if let result = result {
                handler(result)
            }
        }
    }
}
#endif

extension DioRequest: ReactiveCompatible {
}

public extension Reactive where Base: DioDataRequest {
    // MARK: Defaults
    
    /**
     Transform a request into an observable of the response
     
     - parameter queue: The dispatch queue to use.
     - returns: The observable of `NSHTTPURLResponse` for the created request.
     */
    func response(queue: DispatchQueue = .main) -> Observable<HTTPURLResponse> {
        Observable.create { observer in
            let dataRequest = self.base
                .response(queue: queue) { (packedResponse) -> Void in
                    switch packedResponse.result {
                    case .success:
                        if let httpResponse = packedResponse.response {
                            observer.on(.next(httpResponse))
                            observer.on(.completed)
                        } else {
                            observer.on(.error(RxDioUnknownError))
                        }
                    case let .failure(error):
                        observer.on(.error(error as Error))
                    }
                }
            return Disposables.create {
                dataRequest.cancel()
            }
        }
    }
    
    /**
     Transform a request into an observable of the response and serialized object.
     
     - parameter queue: The dispatch queue to use.
     - parameter responseSerializer: The the serializer.
     - returns: The observable of `(NSHTTPURLResponse, T.SerializedObject)` for the created download request.
     */
    func responseResult<T: DataResponseSerializerProtocol>(queue: DispatchQueue = .main, responseSerializer: T) -> Observable<(HTTPURLResponse, T.SerializedObject)> {
        Observable.create { observer in
            let dataRequest = self.base
                .response(queue: queue, responseSerializer: responseSerializer) { (packedResponse) -> Void in
                    switch packedResponse.result {
                    case let .success(result):
                        if let httpResponse = packedResponse.response {
                            observer.on(.next((httpResponse, result)))
                            observer.on(.completed)
                        } else {
                            observer.on(.error(RxDioUnknownError))
                        }
                    case let .failure(error):
                        observer.on(.error(error as Error))
                    }
                }
            return Disposables.create {
                dataRequest.cancel()
            }
        }
    }
    
    @available(*, deprecated, message: "responseJSON deprecated and will be removed in Dio 6. Use responseDecodable instead.")
    func responseJSON() -> Observable<DioDataResponse<Any>> {
        Observable.create { observer in
            let request = self.base
            
            request.responseJSON { response in
                switch response.result {
                case .success:
                    observer.on(.next(response))
                    observer.on(.completed)
                case let .failure(error):
                    observer.on(.error(error))
                }
            }
            
            return Disposables.create {
                request.cancel()
            }
        }
    }
    
    /**
     Transform a request into an observable of the serialized object.
     
     - parameter queue: The dispatch queue to use.
     - parameter responseSerializer: The the serializer.
     - returns: The observable of `T.SerializedObject` for the created download request.
     */
    func result<T: DataResponseSerializerProtocol>(queue: DispatchQueue = .main, responseSerializer: T) -> Observable<T.SerializedObject> {
        Observable.create { observer in
            let dataRequest = self.base
                .response(queue: queue, responseSerializer: responseSerializer) { (packedResponse) -> Void in
                    switch packedResponse.result {
                    case let .success(result):
                        if packedResponse.response != nil {
                            observer.on(.next(result))
                            observer.on(.completed)
                        } else {
                            observer.on(.error(RxDioUnknownError))
                        }
                    case let .failure(error):
                        observer.on(.error(error as Error))
                    }
                }
            return Disposables.create {
                dataRequest.cancel()
            }
        }
    }
    
    /**
     Returns an `Observable` of NSData for the current request.
     
     - parameter cancelOnDispose: Indicates if the request has to be canceled when the observer is disposed, **default:** `false`
     
     - returns: An instance of `Observable<NSData>`
     */
    func responseData() -> Observable<(HTTPURLResponse, Data)> {
        responseResult(responseSerializer: DataResponseSerializer())
    }
    
    func data() -> Observable<Data> {
        result(responseSerializer: DataResponseSerializer())
    }
    
    /**
     Returns an `Observable` of a String for the current request
     
     - parameter encoding: Type of the string encoding, **default:** `nil`
     
     - returns: An instance of `Observable<String>`
     */
    func responseString(encoding: String.Encoding? = nil) -> Observable<(HTTPURLResponse, String)> {
        responseResult(responseSerializer: StringResponseSerializer(encoding: encoding))
    }
    
    func string(encoding: String.Encoding? = nil) -> Observable<String> {
        result(responseSerializer: StringResponseSerializer(encoding: encoding))
    }
    
    /**
     Returns an `Observable` of a serialized JSON for the current request.
     
     - parameter options: Reading options for JSON decoding process, **default:** `.AllowFragments`
     
     - returns: An instance of `Observable<AnyObject>`
     */
    @available(*, deprecated, message: "responseJSON deprecated and will be removed in Dio 6. Use responseDecodable instead.")
    func responseJSON(options: JSONSerialization.ReadingOptions = .allowFragments) -> Observable<(HTTPURLResponse, Any)> {
        responseResult(responseSerializer: JSONResponseSerializer(options: options))
    }
    
    /**
     Returns an `Observable` of a serialized JSON for the current request.
     
     - parameter options: Reading options for JSON decoding process, **default:** `.AllowFragments`
     
     - returns: An instance of `Observable<AnyObject>`
     */
    @available(*, deprecated, message: "json deprecated and will be removed in Dio 6. Use decodable instead.")
    func json(options: JSONSerialization.ReadingOptions = .allowFragments) -> Observable<Any> {
        result(responseSerializer: JSONResponseSerializer(options: options))
    }
    
    /**
     Returns an `Observable` of a serialized Decodable for the current request.
     
     - parameter decoder: The `DataDecoder`. `JSONDecoder()` by default.
     
     - returns: An instance of `Observable<(HTTPURLResponse, T)>`
     */
    func responseDecodable<T: Decodable>(of type: T.Type = T.self,
                                         decoder: DioDataDecoder = JSONDecoder()) -> Observable<(HTTPURLResponse, T)> {
        responseResult(responseSerializer: DecodableResponseSerializer(decoder: decoder))
    }
    
    /**
     Returns an `Observable` of a serialized Decodable for the current request.
     
     - parameter decoder: The `DataDecoder`. `JSONDecoder()` by default.
     
     - returns: An instance of `Observable<T>`
     */
    func decodable<T: Decodable>(of type: T.Type = T.self,
                                 decoder: DioDataDecoder = JSONDecoder()) -> Observable<T> {
        result(responseSerializer: DecodableResponseSerializer(decoder: decoder))
    }
}

#if swift(>=5.5)
public extension Reactive where Base: DioDataRequest {
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    func response(queue: DispatchQueue = .main) async throws -> HTTPURLResponse {
        let packedResponse = await base.response(queue: queue)
        switch packedResponse.result {
        case .success:
            if let httpResponse = packedResponse.response {
                return httpResponse
            } else {
                throw RxDioUnknownError
            }
        case let .failure(error):
            throw error as Error
        }
    }
    
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    @available(*, deprecated, message: "responseJSON deprecated and will be removed in Dio 6. Use responseDecodable instead.")
    func responseJSON() async throws -> DioDataResponse<Any> {
        let response = await base.responseJSON()
        switch response.result {
        case .success:
            return response
        case let .failure(error):
            throw error
        }
    }
    
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    func responseResult<T: DataResponseSerializerProtocol>(queue: DispatchQueue = .main, responseSerializer: T) async throws -> (HTTPURLResponse, T.SerializedObject) {
        let packedResponse = await base.response(queue: queue, responseSerializer: responseSerializer)
        switch packedResponse.result {
        case let .success(result):
            if let httpResponse = packedResponse.response {
                return (httpResponse, result)
            } else {
                throw RxDioUnknownError
            }
        case let .failure(error):
            throw error as Error
        }
    }
    
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    func result<T: DataResponseSerializerProtocol>(queue: DispatchQueue = .main, responseSerializer: T) async throws -> T.SerializedObject {
        let packedResponse = await base.response(queue: queue, responseSerializer: responseSerializer)
        switch packedResponse.result {
        case let .success(result):
            if packedResponse.response != nil {
                return result
            } else {
                throw RxDioUnknownError
            }
        case let .failure(error):
            throw error as Error
        }
    }
    
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    func responseData() async throws -> (HTTPURLResponse, Data) {
        try await responseResult(responseSerializer: DataResponseSerializer())
    }
    
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    func data() async throws -> Data {
        try await result(responseSerializer: DataResponseSerializer())
    }
    
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    func responseString(encoding: String.Encoding? = nil) async throws -> (HTTPURLResponse, String) {
        try await responseResult(responseSerializer: StringResponseSerializer(encoding: encoding))
    }
    
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    func string(encoding: String.Encoding? = nil) async throws -> String {
        try await result(responseSerializer: StringResponseSerializer(encoding: encoding))
    }
    
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    @available(*, deprecated, message: "responseJSON deprecated and will be removed in Dio 6. Use responseDecodable instead.")
    func responseJSON(options: JSONSerialization.ReadingOptions = .allowFragments) async throws -> (HTTPURLResponse, Any) {
        try await responseResult(responseSerializer: JSONResponseSerializer(options: options))
    }
    
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    @available(*, deprecated, message: "json deprecated and will be removed in Dio 6. Use decodable instead.")
    func json(options: JSONSerialization.ReadingOptions = .allowFragments) async throws -> Any {
        try await result(responseSerializer: JSONResponseSerializer(options: options))
    }
    
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    func responseDecodable<T: Decodable>(
        of type: T.Type = T.self,
        decoder: DioDataDecoder = JSONDecoder()
    ) async throws -> (HTTPURLResponse, T) {
        try await responseResult(responseSerializer: DecodableResponseSerializer(decoder: decoder))
    }
    
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    func decodable<T: Decodable>(
        of type: T.Type = T.self,
        decoder: DioDataDecoder = JSONDecoder()
    ) async throws -> T {
        try await result(responseSerializer: DecodableResponseSerializer(decoder: decoder))
    }
}
#endif

public extension Reactive where Base: DioDownloadRequest {
    // MARK: Defaults
    
    func response(queue: DispatchQueue = .main) -> Observable<HTTPURLResponse> {
        Observable.create { observer in
            let dataRequest = self.base
                .response(queue: queue) { (packedResponse) -> Void in
                    switch packedResponse.result {
                    case .success:
                        if let httpResponse = packedResponse.response {
                            observer.on(.next(httpResponse))
                            observer.on(.completed)
                        } else {
                            observer.on(.error(RxDioUnknownError))
                        }
                    case let .failure(error):
                        observer.on(.error(error as Error))
                    }
                }
            return Disposables.create {
                dataRequest.cancel()
            }
        }
    }
    
    func responseResult<T: DownloadResponseSerializerProtocol>(queue: DispatchQueue = .main, responseSerializer: T) -> Observable<(HTTPURLResponse, T.SerializedObject)> {
        Observable.create { observer in
            let dataRequest = self.base
                .response(queue: queue, responseSerializer: responseSerializer) { (packedResponse) -> Void in
                    switch packedResponse.result {
                    case let .success(result):
                        if let httpResponse = packedResponse.response {
                            observer.on(.next((httpResponse, result)))
                            observer.on(.completed)
                        } else {
                            observer.on(.error(RxDioUnknownError))
                        }
                    case let .failure(error):
                        observer.on(.error(error as Error))
                    }
                }
            return Disposables.create {
                dataRequest.cancel()
            }
        }
    }
    
    @available(*, deprecated, message: "responseJSON deprecated and will be removed in Dio 6. Use responseDecodable instead.")
    func responseJSON() -> Observable<DioDownloadResponse<Any>> {
        Observable.create { observer in
            let request = self.base
            
            request.responseJSON { response in
                switch response.result {
                case .success:
                    observer.on(.next(response))
                    observer.on(.completed)
                case let .failure(error):
                    observer.on(.error(error))
                }
            }
            
            return Disposables.create {
                request.cancel()
            }
        }
    }
    
    func result<T: DownloadResponseSerializerProtocol>(queue: DispatchQueue = .main, responseSerializer: T) -> Observable<T.SerializedObject> {
        Observable.create { observer in
            let dataRequest = self.base
                .response(queue: queue, responseSerializer: responseSerializer) { (packedResponse) -> Void in
                    switch packedResponse.result {
                    case let .success(result):
                        if packedResponse.response != nil {
                            observer.on(.next(result))
                            observer.on(.completed)
                        } else {
                            observer.on(.error(RxDioUnknownError))
                        }
                    case let .failure(error):
                        observer.on(.error(error as Error))
                    }
                }
            return Disposables.create {
                dataRequest.cancel()
            }
        }
    }
    
    func responseData() -> Observable<(HTTPURLResponse, Data)> {
        responseResult(responseSerializer: DataResponseSerializer())
    }
    
    func data() -> Observable<Data> {
        result(responseSerializer: DataResponseSerializer())
    }
    
    func responseString(encoding: String.Encoding? = nil) -> Observable<(HTTPURLResponse, String)> {
        responseResult(responseSerializer: StringResponseSerializer(encoding: encoding))
    }
    
    func string(encoding: String.Encoding? = nil) -> Observable<String> {
        result(responseSerializer: StringResponseSerializer(encoding: encoding))
    }
    
    @available(*, deprecated, message: "responseJSON deprecated and will be removed in Dio 6. Use responseDecodable instead.")
    func responseJSON(options: JSONSerialization.ReadingOptions = .allowFragments) -> Observable<(HTTPURLResponse, Any)> {
        responseResult(responseSerializer: JSONResponseSerializer(options: options))
    }
    
    @available(*, deprecated, message: "json deprecated and will be removed in Dio 6. Use decodable instead.")
    func json(options: JSONSerialization.ReadingOptions = .allowFragments) -> Observable<Any> {
        result(responseSerializer: JSONResponseSerializer(options: options))
    }
    
    func responseDecodable<T: Decodable>(
        of type: T.Type = T.self,
        decoder: DioDataDecoder = JSONDecoder()
    ) -> Observable<(HTTPURLResponse, T)> {
        responseResult(responseSerializer: DecodableResponseSerializer(decoder: decoder))
    }
    
    func decodable<T: Decodable>(
        of type: T.Type = T.self,
        decoder: DioDataDecoder = JSONDecoder()
    ) -> Observable<T> {
        result(responseSerializer: DecodableResponseSerializer(decoder: decoder))
    }
}

#if swift(>=5.5)
public extension Reactive where Base: DioDownloadRequest {
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    func response(queue: DispatchQueue = .main) async throws -> HTTPURLResponse {
        let packedResponse = await base.response(queue: queue)
        switch packedResponse.result {
        case .success:
            if let httpResponse = packedResponse.response {
                return httpResponse
            } else {
                throw RxDioUnknownError
            }
        case let .failure(error):
            throw error as Error
        }
    }
    
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    @available(*, deprecated, message: "responseJSON deprecated and will be removed in Dio 6. Use responseDecodable instead.")
    func responseJSON() async throws -> DioDownloadResponse<Any> {
        let response = await base.responseJSON()
        switch response.result {
        case .success:
            return response
        case let .failure(error):
            throw error
        }
    }
    
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    func responseResult<T: DownloadResponseSerializerProtocol>(queue: DispatchQueue = .main, responseSerializer: T) async throws -> (HTTPURLResponse, T.SerializedObject) {
        let packedResponse = await base.response(queue: queue, responseSerializer: responseSerializer)
        switch packedResponse.result {
        case let .success(result):
            if let httpResponse = packedResponse.response {
                return (httpResponse, result)
            } else {
                throw RxDioUnknownError
            }
        case let .failure(error):
            throw error as Error
        }
    }
    
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    func result<T: DownloadResponseSerializerProtocol>(queue: DispatchQueue = .main, responseSerializer: T) async throws -> T.SerializedObject {
        let packedResponse = await base.response(queue: queue, responseSerializer: responseSerializer)
        switch packedResponse.result {
        case let .success(result):
            if packedResponse.response != nil {
                return result
            } else {
                throw RxDioUnknownError
            }
        case let .failure(error):
            throw error as Error
        }
    }
    
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    func responseData() async throws -> (HTTPURLResponse, Data) {
        try await responseResult(responseSerializer: DataResponseSerializer())
    }
    
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    func data() async throws -> Data {
        try await result(responseSerializer: DataResponseSerializer())
    }
    
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    func responseString(encoding: String.Encoding? = nil) async throws -> (HTTPURLResponse, String) {
        try await responseResult(responseSerializer: StringResponseSerializer(encoding: encoding))
    }
    
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    func string(encoding: String.Encoding? = nil) async throws -> String {
        try await result(responseSerializer: StringResponseSerializer(encoding: encoding))
    }
    
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    @available(*, deprecated, message: "responseJSON deprecated and will be removed in Dio 6. Use responseDecodable instead.")
    func responseJSON(options: JSONSerialization.ReadingOptions = .allowFragments) async throws -> (HTTPURLResponse, Any) {
        try await responseResult(responseSerializer: JSONResponseSerializer(options: options))
    }
    
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    @available(*, deprecated, message: "json deprecated and will be removed in Dio 6. Use decodable instead.")
    func json(options: JSONSerialization.ReadingOptions = .allowFragments) async throws -> Any {
        try await result(responseSerializer: JSONResponseSerializer(options: options))
    }
    
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    func responseDecodable<T: Decodable>(
        of type: T.Type = T.self,
        decoder: DioDataDecoder = JSONDecoder()
    ) async throws -> (HTTPURLResponse, T) {
        try await responseResult(responseSerializer: DecodableResponseSerializer(decoder: decoder))
    }
    
    @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
    func decodable<T: Decodable>(
        of type: T.Type = T.self,
        decoder: DioDataDecoder = JSONDecoder()
    ) async throws -> T {
        try await result(responseSerializer: DecodableResponseSerializer(decoder: decoder))
    }
}
#endif

public extension Reactive where Base: DioRequest {
    // MARK: Request - Upload and download progress
    
    /**
     Returns an `Observable` for the current progress status.
     
     Parameters on observed tuple:
     
     1. bytes written so far.
     1. total bytes to write.
     
     - returns: An instance of `Observable<RxProgress>`
     */
    func progress() -> Observable<RxProgress> {
        Observable.create { observer in
            let handler: DioRequest.ProgressHandler = { progress in
                let rxProgress = RxProgress(bytesWritten: progress.completedUnitCount, totalBytes: progress.totalUnitCount)
                observer.on(.next(rxProgress))
                
                if rxProgress.bytesWritten >= rxProgress.totalBytes {
                    observer.on(.completed)
                }
            }
            
            // Try in following order:
            //  - UploadRequest (Inherits from DataRequest, so we test the discrete case first)
            //  - DownloadRequest
            //  - DataRequest
            if let uploadReq = self.base as? DioUploadRequest {
                uploadReq.uploadProgress(closure: handler)
            } else if let downloadReq = self.base as? DioDownloadRequest {
                downloadReq.downloadProgress(closure: handler)
            } else if let dataReq = self.base as? DioDataRequest {
                dataReq.downloadProgress(closure: handler)
            }
            
            return Disposables.create()
        }
        // warm up a bit :)
        .startWith(RxProgress(bytesWritten: 0, totalBytes: 0))
    }
}

// MARK: RxProgress
public struct RxProgress {
    public let bytesWritten: Int64
    public let totalBytes: Int64
    
    public init(bytesWritten: Int64, totalBytes: Int64) {
        self.bytesWritten = bytesWritten
        self.totalBytes = totalBytes
    }
}

public extension RxProgress {
    var bytesRemaining: Int64 {
        totalBytes - bytesWritten
    }
    
    var completed: Float {
        guard totalBytes > 0 else {
            return 0
        }
        return Float(bytesWritten) / Float(totalBytes)
    }
}

extension RxProgress: Equatable {
}

public func ==(lhs: RxProgress, rhs: RxProgress) -> Bool {
    lhs.bytesWritten == rhs.bytesWritten && lhs.totalBytes == rhs.totalBytes
}
